package io.gitee.sections.sequence.core.impl;

import io.gitee.sections.sequence.core.SequenceDefinition;
import io.gitee.sections.sequence.core.cache.CacheProvider;
import io.gitee.sections.sequence.core.dao.DataAccessor;
import io.gitee.sections.sequence.core.exp.SequenceException;

import java.util.List;
import java.util.Optional;

public class Sequence {
    private SequenceDefinition definition;
    private CacheProvider cacheProvider;
    private DataAccessor dataAccessor;

    public Sequence(SequenceDefinition definition, List<CacheProvider> cacheProviders) {
        this.definition = definition.copy();
        this.cacheProvider = cacheProviders.stream()
                .filter(p -> p.support(definition.getCacheMode()))
                .findFirst()
                .orElseThrow(() -> {
                    String msg = String.format("no cache provider support cache mode: %s", definition.getCacheMode());
                    return new IllegalArgumentException(msg);
                });
    }

    public void initialize(DataAccessor dataAccessor) {
        this.dataAccessor = dataAccessor;
        try {
            doInitialize();
        } catch (Exception e) {
            String msg = String.format("fail to initialize sequence %s", definition.toString());
            throw new SequenceException(msg, e);
        } finally {
            this.dataAccessor.close();
        }
    }

    /**
     * 数据库里不存在该配置项则插入到数据库中
     * 数据中已经存在，则更新数据库中的配置
     */
    private synchronized void doInitialize() {
        if (!dataAccessor.find(definition.getKey()).isPresent()) {
            dataAccessor.insert(definition);
        } else {
            dataAccessor.update(definition);
        }
        reloadCache();
    }

    private void reloadCache() {
        long newNumber = dataAccessor.grow(definition);
        for (int i = 0; i < definition.getCacheSize(); i++) {
            final long number = newNumber - (definition.getCacheSize() - i) * definition.getStepSize();
            cacheProvider.add(definition.getKey(), number);
        }
    }

    public Long next() {
        try {
            return doGetNext();
        } catch (Exception e) {
            String msg = String.format("fail to get next of sequence %s", definition.toString());
            throw new SequenceException(msg, e);
        } finally {
            this.dataAccessor.close();
        }
    }

    private synchronized Long doGetNext() {
        Optional<Long> number = cacheProvider.poll(definition.getKey());
        if (!number.isPresent()) {
            reloadCache();
            number = cacheProvider.poll(definition.getKey());
        }
        return number.get();
    }

    public SequenceDefinition getDefinition() {
        return definition.copy();
    }
}
